
#ifndef _PROCESSROUTER_CHANNEL_H_
#define _PROCESSROUTER_CHANNEL_H_

#include "message.h"
#include "iothread.h"
#include <czmq.h>

namespace ProcessRouter {

class Channel;

class ChannelHandle: public MessageProtocol {
    public:
        ChannelHandle();

        void setChannel(Channel *channel);
        Channel *channel();

        virtual void didRecevied() {}
        virtual void didConnected() {}
        virtual void didError() {}

    private:
        Channel *m_channel;
};

struct ZAddress {
    public:
        ZAddress(byte *addr, size_t size)
        {
            m_size = size;
            m_addr = (byte*)malloc(size);
            memcpy(m_addr, addr, size);
            toString();
        }

        ~ZAddress()
        {
            free(m_addr);
        }

        std::string str() { return m_str; }

        byte *m_addr;
        size_t m_size;

    private:
        void toString();
        std::string m_str;
};

class CMessage: public Message {
    public:
        CMessage(std::string content)
            : Message(content)
        {
        }

        CMessage()
            : Message()
        {
        }

        CMessage(const CMessage& zm)
        {
            m_address = zm.m_address;
            m_isNull = false;
            m_content = zm.m_content;
        }

        std::string addressString();

        byte* address();
        void setAddr(byte* addr, size_t size);

        boost::shared_ptr<ZAddress> m_address;
};

class ChildProceeErrorEvent : public Rooftop::EventBase {
    public:
        ChildProceeErrorEvent(int cpid)
            : Rooftop::EventBase(CHILDPROCESSERROREVENT)
            , m_cpid(cpid)
        {}

        int cpid() { return m_cpid; }
    private:
        int m_cpid;
};

class CMessageEvent : public Rooftop::EventBase {
    public:
        enum Type{
            ZMQTYPE = 1,
            FIFOTYPE = 2,
        };

        CMessageEvent(CMessage msg, Type type);
        ~CMessageEvent() {}

        CMessage message() { return m_message; }

        void setFromPid(int pid) { m_pid = pid; }

        Type type() { return m_type; }

        int fromPid() { return m_pid; }

    private:
        CMessage m_message;
        int m_pid;
        Type m_type;
};

class Channel: public Listener {
    public:
        Channel(IOThread *thread);
        ~Channel();

        virtual void open() {}
        virtual CMessage recv() = 0;
        virtual int send(CMessage &message) = 0;

        void setClient(ChannelHandle *client);

        // call it by user
        virtual void destroy() {}
};

class FifoChannel: public Channel {
    public:
        FifoChannel(IOThread *thread);
        ~FifoChannel();

        virtual void open();
        virtual CMessage recv();
        virtual int send(CMessage &message);

        void setChild();
        void forkInit(ChannelHandle *client);
        int readFd();
        int writeFd();

        int cfork();

        int pid() { return m_pid; }
        int childPid() { return m_childPid; }
        void setChildPid(int pid) { m_childPid = pid; }

        virtual void destroy();

    private:
        int m_pfds[2];
        int m_cfds[2];
        bool m_isChild;
        int m_childPid;
        int m_pid;
        ev_child m_cw;
};

class ZmqChannel: public Channel {
    public:
        ZmqChannel(IOThread *thread);
        ~ZmqChannel();

        void setProtocol(std::string protocol);
        std::string protocol();

        virtual void open();
        virtual CMessage recv();
        virtual int send(CMessage &message);

        virtual void destroy();

    private:
        void getAddress(zmsg_t* msg, byte** _addr, size_t *_size);
        std::string getMessage(zmsg_t* msg);

        zctx_t *m_zmqctx;
        void *m_zmqsocket;
        std::string m_protocol;
};

}

#endif

